# Copyright 2017 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import collections
import io
import subprocess
import unittest
from unittest import mock

from mobly.controllers.android_device_lib import adb

# Mock parameters for instrumentation.
MOCK_INSTRUMENTATION_PACKAGE = 'com.my.instrumentation.tests'
MOCK_INSTRUMENTATION_RUNNER = 'com.my.instrumentation.runner'
MOCK_INSTRUMENTATION_OPTIONS = collections.OrderedDict([
    ('option1', 'value1'),
    ('option2', 'value2'),
])
# Mock android instrumentation commands.
MOCK_BASIC_INSTRUMENTATION_COMMAND = (
    'am instrument -r -w  com.my'
    '.instrumentation.tests/com.android'
    '.common.support.test.runner'
    '.AndroidJUnitRunner'
)
MOCK_RUNNER_INSTRUMENTATION_COMMAND = (
    'am instrument -r -w  com.my'
    '.instrumentation.tests/com.my'
    '.instrumentation.runner'
)
MOCK_OPTIONS_INSTRUMENTATION_COMMAND = (
    'am instrument -r -w -e option1 value1'
    ' -e option2 value2 com.my'
    '.instrumentation.tests/com.android'
    '.common.support.test.runner'
    '.AndroidJUnitRunner'
)

# Mock root command outputs.
MOCK_ROOT_SUCCESS_OUTPUT = 'adbd is already running as root'
MOCK_ROOT_ERROR_OUTPUT = 'adb: unable to connect for root: closed'.encode(
    'utf-8'
)

# Mock Shell Command
MOCK_SHELL_COMMAND = 'ls'
MOCK_COMMAND_OUTPUT = '/system/bin/ls'.encode('utf-8')
MOCK_DEFAULT_STDOUT = 'out'
MOCK_DEFAULT_STDERR = 'err'
MOCK_DEFAULT_COMMAND_OUTPUT = MOCK_DEFAULT_STDOUT.encode('utf-8')
MOCK_ADB_SHELL_COMMAND_CHECK = 'adb shell command -v ls'


class AdbTest(unittest.TestCase):
  """Unit tests for mobly.controllers.android_device_lib.adb."""

  def _mock_execute_and_process_stdout_process(self, mock_popen):
    # the created proc object in adb._execute_and_process_stdout()
    mock_proc = mock.Mock()
    mock_popen.return_value = mock_proc

    mock_popen.return_value.stdout.readline.side_effect = ['']

    mock_proc.communicate = mock.Mock(
        return_value=('', MOCK_DEFAULT_STDERR.encode('utf-8'))
    )
    mock_proc.returncode = 0
    return mock_popen

  @mock.patch('mobly.utils.run_command')
  def test_is_adb_available(self, mock_run_command):
    mock_run_command.return_value = (
        0,
        '/usr/local/bin/adb\n'.encode('utf-8'),
        ''.encode('utf-8'),
    )
    self.assertTrue(adb.is_adb_available())

  @mock.patch('mobly.utils.run_command')
  def test_is_adb_available_negative(self, mock_run_command):
    mock_run_command.return_value = (0, ''.encode('utf-8'), ''.encode('utf-8'))
    self.assertFalse(adb.is_adb_available())

  @mock.patch('mobly.utils.run_command')
  def test_exec_cmd_no_timeout_success(self, mock_run_command):
    mock_run_command.return_value = (
        0,
        MOCK_DEFAULT_STDOUT.encode('utf-8'),
        MOCK_DEFAULT_STDERR.encode('utf-8'),
    )
    out = adb.AdbProxy()._exec_cmd(
        ['fake_cmd'], shell=False, timeout=None, stderr=None
    )
    self.assertEqual(MOCK_DEFAULT_STDOUT, out.decode('utf-8'))
    mock_run_command.assert_called_with(['fake_cmd'], shell=False, timeout=None)

  @mock.patch('mobly.utils.run_command')
  def test_exec_cmd_error_with_serial(self, mock_run_command):
    # Return 1 for retcode for error.
    mock_run_command.return_value = (
        1,
        MOCK_DEFAULT_STDOUT.encode('utf-8'),
        MOCK_DEFAULT_STDERR.encode('utf-8'),
    )
    mock_serial = 'ABCD1234'
    with self.assertRaisesRegex(
        adb.AdbError, 'Error executing adb cmd .*'
    ) as context:
      adb.AdbProxy(mock_serial).fake_cmd()
    self.assertEqual(context.exception.serial, mock_serial)
    self.assertIn(mock_serial, context.exception.cmd)

  @mock.patch('mobly.utils.run_command')
  def test_exec_cmd_error_without_serial(self, mock_run_command):
    # Return 1 for retcode for error.
    mock_run_command.return_value = (
        1,
        MOCK_DEFAULT_STDOUT.encode('utf-8'),
        MOCK_DEFAULT_STDERR.encode('utf-8'),
    )
    with self.assertRaisesRegex(
        adb.AdbError, 'Error executing adb cmd .*'
    ) as context:
      adb.AdbProxy()._exec_cmd(
          ['fake_cmd'], shell=False, timeout=None, stderr=None
      )
    self.assertFalse(context.exception.serial)
    mock_run_command.assert_called_with(['fake_cmd'], shell=False, timeout=None)

  @mock.patch('mobly.utils.run_command')
  def test_exec_cmd_with_timeout_success(self, mock_run_command):
    mock_run_command.return_value = (
        0,
        MOCK_DEFAULT_STDOUT.encode('utf-8'),
        MOCK_DEFAULT_STDERR.encode('utf-8'),
    )

    out = adb.AdbProxy()._exec_cmd(
        ['fake_cmd'], shell=False, timeout=1, stderr=None
    )
    self.assertEqual(MOCK_DEFAULT_STDOUT, out.decode('utf-8'))
    mock_run_command.assert_called_with(['fake_cmd'], shell=False, timeout=1)

  @mock.patch('mobly.utils.run_command')
  def test_exec_cmd_timed_out(self, mock_run_command):
    mock_run_command.side_effect = subprocess.TimeoutExpired(
        cmd='mock_command', timeout=0.01
    )
    mock_serial = '1234Abcd'

    with self.assertRaisesRegex(
        adb.AdbTimeoutError,
        'Timed out executing command "adb -s 1234Abcd fake-cmd" after 0.01s.',
    ) as context:
      adb.AdbProxy(mock_serial).fake_cmd(timeout=0.01)

    self.assertEqual(context.exception.serial, mock_serial)
    self.assertIn(mock_serial, context.exception.cmd)

  @mock.patch('mobly.utils.run_command')
  def test_exec_cmd_timed_out_without_serial(self, mock_run_command):
    mock_run_command.side_effect = subprocess.TimeoutExpired(
        cmd='mock_command', timeout=0.01
    )

    with self.assertRaisesRegex(
        adb.AdbTimeoutError,
        'Timed out executing command "adb fake-cmd" after 0.01s.',
    ):
      adb.AdbProxy().fake_cmd(timeout=0.01)

  def test_exec_cmd_with_negative_timeout_value(self):
    with self.assertRaisesRegex(
        ValueError, 'Timeout is not a positive value: -1'
    ):
      adb.AdbProxy()._exec_cmd(
          ['fake_cmd'], shell=False, timeout=-1, stderr=None
      )

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  def test_execute_and_process_stdout_reads_stdout(self, mock_popen):
    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_popen.return_value.stdout.readline.side_effect = ['1', '2', '']
    mock_handler = mock.MagicMock()

    err = adb.AdbProxy()._execute_and_process_stdout(
        ['fake_cmd'], shell=False, handler=mock_handler
    )
    self.assertEqual(mock_handler.call_count, 2)
    mock_handler.assert_any_call('1')
    mock_handler.assert_any_call('2')

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  def test_execute_and_process_stdout_reads_unexpected_stdout(self, mock_popen):
    unexpected_stdout = MOCK_DEFAULT_STDOUT.encode('utf-8')

    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_handler = mock.MagicMock()
    mock_popen.return_value.communicate = mock.Mock(
        return_value=(unexpected_stdout, MOCK_DEFAULT_STDERR.encode('utf-8'))
    )

    err = adb.AdbProxy()._execute_and_process_stdout(
        ['fake_cmd'], shell=False, handler=mock_handler
    )
    self.assertEqual(mock_handler.call_count, 1)
    mock_handler.assert_called_with(unexpected_stdout)

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  @mock.patch('logging.debug')
  def test_execute_and_process_stdout_logs_cmd(
      self, mock_debug_logger, mock_popen
  ):
    raw_expected_stdout = ''
    expected_stdout = '[elided, processed via handler]'
    expected_stderr = MOCK_DEFAULT_STDERR.encode('utf-8')
    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_popen.return_value.communicate = mock.Mock(
        return_value=(raw_expected_stdout, expected_stderr)
    )

    err = adb.AdbProxy()._execute_and_process_stdout(
        ['fake_cmd'], shell=False, handler=mock.MagicMock()
    )
    mock_debug_logger.assert_called_with(
        'cmd: %s, stdout: %s, stderr: %s, ret: %s',
        'fake_cmd',
        expected_stdout,
        expected_stderr,
        0,
    )

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  @mock.patch('logging.debug')
  def test_execute_and_process_stdout_logs_cmd_with_unexpected_stdout(
      self, mock_debug_logger, mock_popen
  ):
    raw_expected_stdout = MOCK_DEFAULT_STDOUT.encode('utf-8')
    expected_stdout = '[unexpected stdout] %s' % raw_expected_stdout
    expected_stderr = MOCK_DEFAULT_STDERR.encode('utf-8')

    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_popen.return_value.communicate = mock.Mock(
        return_value=(raw_expected_stdout, expected_stderr)
    )

    err = adb.AdbProxy()._execute_and_process_stdout(
        ['fake_cmd'], shell=False, handler=mock.MagicMock()
    )
    mock_debug_logger.assert_called_with(
        'cmd: %s, stdout: %s, stderr: %s, ret: %s',
        'fake_cmd',
        expected_stdout,
        expected_stderr,
        0,
    )

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  def test_execute_and_process_stdout_despite_cmd_exits(self, mock_popen):
    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_popen.return_value.poll.side_effect = [None, 0]
    mock_popen.return_value.stdout.readline.side_effect = ['1', '2', '3', '']
    mock_handler = mock.MagicMock()

    err = adb.AdbProxy()._execute_and_process_stdout(
        ['fake_cmd'], shell=False, handler=mock_handler
    )

    self.assertEqual(mock_handler.call_count, 3)
    mock_handler.assert_any_call('1')
    mock_handler.assert_any_call('2')
    mock_handler.assert_any_call('3')

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  def test_execute_and_process_stdout_when_cmd_eof(self, mock_popen):
    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_popen.return_value.stdout.readline.side_effect = ['1', '2', '3', '']
    mock_handler = mock.MagicMock()

    err = adb.AdbProxy()._execute_and_process_stdout(
        ['fake_cmd'], shell=False, handler=mock_handler
    )

    self.assertEqual(mock_handler.call_count, 3)
    mock_handler.assert_any_call('1')
    mock_handler.assert_any_call('2')
    mock_handler.assert_any_call('3')

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  def test_execute_and_process_stdout_returns_stderr(self, mock_popen):
    self._mock_execute_and_process_stdout_process(mock_popen)

    err = adb.AdbProxy()._execute_and_process_stdout(
        ['fake_cmd'], shell=False, handler=mock.MagicMock()
    )
    self.assertEqual(MOCK_DEFAULT_STDERR, err.decode('utf-8'))

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  def test_execute_and_process_stdout_raises_adb_error(self, mock_popen):
    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_popen.return_value.returncode = 1

    with self.assertRaisesRegex(adb.AdbError, 'Error executing adb cmd .*'):
      err = adb.AdbProxy()._execute_and_process_stdout(
          ['fake_cmd'], shell=False, handler=mock.MagicMock()
      )

  @mock.patch('mobly.controllers.android_device_lib.adb.subprocess.Popen')
  def test_execute_and_process_stdout_when_handler_crash(self, mock_popen):
    self._mock_execute_and_process_stdout_process(mock_popen)
    mock_popen.return_value.stdout.readline.side_effect = ['1', '2', '3', '']
    mock_handler = mock.MagicMock()
    mock_handler.side_effect = ['', TypeError('fake crash'), '', '']

    with self.assertRaisesRegex(TypeError, 'fake crash'):
      err = adb.AdbProxy()._execute_and_process_stdout(
          ['fake_cmd'], shell=False, handler=mock_handler
      )

    mock_popen.return_value.communicate.assert_called_once_with()

  def test_construct_adb_cmd(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd('shell', 'arg1', shell=False)
    self.assertEqual(adb_cmd, ['adb', 'shell', 'arg1'])

  def test_construct_adb_cmd_with_one_command(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd(
        'shell ls /asdafsfd/asdf-asfd/asa', [], shell=False
    )
    self.assertEqual(adb_cmd, ['adb', 'shell ls /asdafsfd/asdf-asfd/asa'])

  def test_construct_adb_cmd_with_one_arg_command(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd(
        'shell', 'ls /asdafsfd/asdf-asfd/asa', shell=False
    )
    self.assertEqual(adb_cmd, ['adb', 'shell', 'ls /asdafsfd/asdf-asfd/asa'])

  def test_construct_adb_cmd_with_one_arg_command_list(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd(
        'shell', ['ls /asdafsfd/asdf-asfd/asa'], shell=False
    )
    self.assertEqual(adb_cmd, ['adb', 'shell', 'ls /asdafsfd/asdf-asfd/asa'])

  def test_construct_adb_cmd_with_special_characters(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd(
        'shell', ['a b', '"blah"', r'\/\/'], shell=False
    )
    self.assertEqual(adb_cmd, ['adb', 'shell', 'a b', '"blah"', r'\/\/'])

  def test_construct_adb_cmd_with_serial(self):
    adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd(
        'shell', 'arg1', shell=False
    )
    self.assertEqual(adb_cmd, ['adb', '-s', '12345', 'shell', 'arg1'])

  def test_construct_adb_cmd_with_list(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd(
        'shell', ['arg1', 'arg2'], shell=False
    )
    self.assertEqual(adb_cmd, ['adb', 'shell', 'arg1', 'arg2'])

  def test_construct_adb_cmd_with_serial_with_list(self):
    adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd(
        'shell', ['arg1', 'arg2'], shell=False
    )
    self.assertEqual(adb_cmd, ['adb', '-s', '12345', 'shell', 'arg1', 'arg2'])

  def test_construct_adb_cmd_with_shell_true(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd(
        'shell', 'arg1 arg2', shell=True
    )
    self.assertEqual(adb_cmd, '"adb" shell arg1 arg2')

  def test_construct_adb_cmd_with_shell_true_with_one_command(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd(
        'shell ls /asdafsfd/asdf-asfd/asa', [], shell=True
    )
    self.assertEqual(adb_cmd, '"adb" shell ls /asdafsfd/asdf-asfd/asa ')

  def test_construct_adb_cmd_with_shell_true_with_one_arg_command(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd(
        'shell', 'ls /asdafsfd/asdf-asfd/asa', shell=True
    )
    self.assertEqual(adb_cmd, '"adb" shell ls /asdafsfd/asdf-asfd/asa')

  def test_construct_adb_cmd_with_shell_true_with_one_arg_command_list(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd(
        'shell', ['ls /asdafsfd/asdf-asfd/asa'], shell=True
    )
    self.assertEqual(adb_cmd, '"adb" shell \'ls /asdafsfd/asdf-asfd/asa\'')

  def test_construct_adb_cmd_with_shell_true_with_auto_quotes(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd(
        'shell', ['a b', '"blah"', r'\/\/'], shell=True
    )
    self.assertEqual(adb_cmd, "\"adb\" shell 'a b' '\"blah\"' '\\/\\/'")

  def test_construct_adb_cmd_with_shell_true_with_serial(self):
    adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd(
        'shell', 'arg1 arg2', shell=True
    )
    self.assertEqual(adb_cmd, '"adb" -s "12345" shell arg1 arg2')

  def test_construct_adb_cmd_with_shell_true_with_list(self):
    adb_cmd = adb.AdbProxy()._construct_adb_cmd(
        'shell', ['arg1', 'arg2'], shell=True
    )
    self.assertEqual(adb_cmd, '"adb" shell arg1 arg2')

  def test_construct_adb_cmd_with_shell_true_with_serial_with_list(self):
    adb_cmd = adb.AdbProxy('12345')._construct_adb_cmd(
        'shell', ['arg1', 'arg2'], shell=True
    )
    self.assertEqual(adb_cmd, '"adb" -s "12345" shell arg1 arg2')

  def test_exec_adb_cmd(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT
      adb.AdbProxy().shell(['arg1', 'arg2'])
      mock_exec_cmd.assert_called_once_with(
          ['adb', 'shell', 'arg1', 'arg2'],
          shell=False,
          timeout=None,
          stderr=None,
      )

  def test_exec_adb_cmd_with_shell_true(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT
      adb.AdbProxy().shell('arg1 arg2', shell=True)
      mock_exec_cmd.assert_called_once_with(
          '"adb" shell arg1 arg2', shell=True, timeout=None, stderr=None
      )

  def test_exec_adb_cmd_formats_command(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      with mock.patch.object(
          adb.AdbProxy, '_construct_adb_cmd'
      ) as mock_construct_adb_cmd:
        mock_adb_cmd = mock.MagicMock()
        mock_adb_args = mock.MagicMock()
        mock_construct_adb_cmd.return_value = mock_adb_cmd
        mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT

        adb.AdbProxy().shell(mock_adb_args)
        mock_construct_adb_cmd.assert_called_once_with(
            'shell', mock_adb_args, shell=False
        )
        mock_exec_cmd.assert_called_once_with(
            mock_adb_cmd, shell=False, timeout=None, stderr=None
        )

  def test_exec_adb_cmd_formats_command_with_shell_true(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      with mock.patch.object(
          adb.AdbProxy, '_construct_adb_cmd'
      ) as mock_construct_adb_cmd:
        mock_adb_cmd = mock.MagicMock()
        mock_adb_args = mock.MagicMock()
        mock_construct_adb_cmd.return_value = mock_adb_cmd

        adb.AdbProxy().shell(mock_adb_args, shell=True)
        mock_construct_adb_cmd.assert_called_once_with(
            'shell', mock_adb_args, shell=True
        )
        mock_exec_cmd.assert_called_once_with(
            mock_adb_cmd, shell=True, timeout=None, stderr=None
        )

  def test_execute_adb_and_process_stdout_formats_command(self):
    with mock.patch.object(
        adb.AdbProxy, '_execute_and_process_stdout'
    ) as mock_execute_and_process_stdout:
      with mock.patch.object(
          adb.AdbProxy, '_construct_adb_cmd'
      ) as mock_construct_adb_cmd:
        mock_adb_cmd = mock.MagicMock()
        mock_adb_args = mock.MagicMock()
        mock_handler = mock.MagicMock()
        mock_construct_adb_cmd.return_value = mock_adb_cmd

        adb.AdbProxy()._execute_adb_and_process_stdout(
            'shell', mock_adb_args, shell=False, handler=mock_handler
        )
        mock_construct_adb_cmd.assert_called_once_with(
            'shell', mock_adb_args, shell=False
        )
        mock_execute_and_process_stdout.assert_called_once_with(
            mock_adb_cmd, shell=False, handler=mock_handler
        )

  @mock.patch('mobly.utils.run_command')
  def test_exec_adb_cmd_with_stderr_pipe(self, mock_run_command):
    mock_run_command.return_value = (
        0,
        MOCK_DEFAULT_STDOUT.encode('utf-8'),
        MOCK_DEFAULT_STDERR.encode('utf-8'),
    )
    stderr_redirect = io.BytesIO()
    out = adb.AdbProxy().shell('arg1 arg2', shell=True, stderr=stderr_redirect)
    self.assertEqual(MOCK_DEFAULT_STDOUT, out.decode('utf-8'))
    self.assertEqual(
        MOCK_DEFAULT_STDERR, stderr_redirect.getvalue().decode('utf-8')
    )

  @mock.patch('mobly.utils.run_command')
  def test_connect_success(self, mock_run_command):
    mock_address = 'localhost:1234'
    mock_run_command.return_value = (
        0,
        f'connected to {mock_address}'.encode('utf-8'),
        MOCK_DEFAULT_STDERR.encode('utf-8'),
    )

    out = adb.AdbProxy().connect(mock_address)
    self.assertEqual('connected to localhost:1234', out.decode('utf-8'))

  @mock.patch('mobly.utils.run_command')
  def test_connect_already_connected(self, mock_run_command):
    mock_address = 'localhost:1234'
    mock_run_command.return_value = (
        0,
        f'already connected to {mock_address}'.encode('utf-8'),
        MOCK_DEFAULT_STDERR.encode('utf-8'),
    )

    out = adb.AdbProxy().connect(mock_address)
    self.assertEqual('already connected to localhost:1234', out.decode('utf-8'))

  @mock.patch('mobly.utils.run_command')
  def test_connect_fail(self, mock_run_command):
    mock_address = 'localhost:1234'
    mock_run_command.return_value = (
        0,
        'Connection refused\n'.encode('utf-8'),
        MOCK_DEFAULT_STDERR.encode('utf-8'),
    )

    with self.assertRaisesRegex(
        adb.AdbError, 'Error executing adb cmd "connect localhost:1234".'
    ):
      out = adb.AdbProxy().connect(mock_address)

  def test_getprop(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = b'blah'
      self.assertEqual(adb.AdbProxy().getprop('haha'), 'blah')
      mock_exec_cmd.assert_called_once_with(
          ['adb', 'shell', 'getprop', 'haha'],
          shell=False,
          stderr=None,
          timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC,
      )

  def test_getprop_custom_timeout(self):
    timeout_s = adb.DEFAULT_GETPROP_TIMEOUT_SEC * 2
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = b'blah'
      self.assertEqual(
          adb.AdbProxy().getprop('haha', timeout=timeout_s), 'blah'
      )
      mock_exec_cmd.assert_called_once_with(
          ['adb', 'shell', 'getprop', 'haha'],
          shell=False,
          stderr=None,
          timeout=timeout_s,
      )

  def test__parse_getprop_output_special_values(self):
    mock_adb_output = (
        b'[selinux.restorecon_recursive]: [/data/misc_ce/10]\n'
        b'[selinux.abc]: [key: value]\n'  # "key: value" as value
        b'[persist.sys.boot.reason.history]: [reboot,adb,1558549857\n'
        b'reboot,factory_reset,1558483886\n'  # multi-line value
        b'reboot,1558483823]\n'
        b'[persist.something]: [haha\n'
        b']\n'
        b'[[wrapped.key]]: [[wrapped value]]\n'
        b'[persist.byte]: [J\xaa\x8bb\xab\x9dP\x0f]\n'  # non-decodable
    )
    parsed_props = adb.AdbProxy()._parse_getprop_output(mock_adb_output)
    expected_output = {
        'persist.sys.boot.reason.history': (
            'reboot,adb,1558549857\nreboot,factory_reset,1558483886\n'
            'reboot,1558483823'
        ),
        'selinux.abc': 'key: value',
        'persist.something': 'haha\n',
        'selinux.restorecon_recursive': '/data/misc_ce/10',
        '[wrapped.key]': '[wrapped value]',
        'persist.byte': 'JbP\x0f',
    }
    self.assertEqual(parsed_props, expected_output)

  def test__parse_getprop_output_malformat_output(self):
    mock_adb_output = (  # Malformat
        b'[selinux.restorecon_recursive][/data/misc_ce/10]\n'
        b'[persist.sys.boot.reason]: [reboot,adb,1558549857]\n'
        b'[persist.something]: [haha]\n'
    )
    parsed_props = adb.AdbProxy()._parse_getprop_output(mock_adb_output)
    expected_output = {
        'persist.sys.boot.reason': 'reboot,adb,1558549857',
        'persist.something': 'haha',
    }
    self.assertEqual(parsed_props, expected_output)

  def test__parse_getprop_output_special_line_separator(self):
    mock_adb_output = (  # Malformat
        b'[selinux.restorecon_recursive][/data/misc_ce/10]\r\n'
        b'[persist.sys.boot.reason]: [reboot,adb,1558549857]\r\n'
        b'[persist.something]: [haha]\r\n'
    )
    parsed_props = adb.AdbProxy()._parse_getprop_output(mock_adb_output)
    expected_output = {
        'persist.sys.boot.reason': 'reboot,adb,1558549857',
        'persist.something': 'haha',
    }
    self.assertEqual(parsed_props, expected_output)

  @mock.patch('time.sleep', return_value=mock.MagicMock())
  def test_getprops(self, mock_sleep):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = (
          b'\n[sendbug.preferred.domain]: [google.com]\n'
          b'[sys.uidcpupower]: []\n'
          b'[sys.wifitracing.started]: [1]\n'
          b'[telephony.lteOnCdmaDevice]: [1]\n\n'
      )
      actual_output = adb.AdbProxy().getprops([
          'sys.wifitracing.started',  # "numeric" value
          'sys.uidcpupower',  # empty value
          'sendbug.preferred.domain',  # string value
          'nonExistentProp',
      ])
      self.assertEqual(
          actual_output,
          {
              'sys.wifitracing.started': '1',
              'sys.uidcpupower': '',
              'sendbug.preferred.domain': 'google.com',
          },
      )
      mock_exec_cmd.assert_called_once_with(
          ['adb', 'shell', 'getprop'],
          shell=False,
          stderr=None,
          timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC,
      )
      mock_sleep.assert_not_called()

  @mock.patch('time.sleep', return_value=mock.MagicMock())
  def test_getprops_when_empty_string_randomly_returned(self, mock_sleep):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.side_effect = [
          b'',
          b'\n[ro.build.id]: [AB42]\n[ro.build.type]: [userdebug]\n\n',
      ]
      actual_output = adb.AdbProxy().getprops(['ro.build.id'])
      self.assertEqual(
          actual_output,
          {
              'ro.build.id': 'AB42',
          },
      )
      self.assertEqual(mock_exec_cmd.call_count, 2)
      mock_exec_cmd.assert_called_with(
          ['adb', 'shell', 'getprop'],
          shell=False,
          stderr=None,
          timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC,
      )
      self.assertEqual(mock_sleep.call_count, 1)
      mock_sleep.assert_called_with(1)

  @mock.patch('time.sleep', return_value=mock.MagicMock())
  def test_getprops_when_empty_string_always_returned(self, mock_sleep):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = b''
      actual_output = adb.AdbProxy().getprops(['ro.build.id'])
      self.assertEqual(actual_output, {})
      self.assertEqual(mock_exec_cmd.call_count, 3)
      mock_exec_cmd.assert_called_with(
          ['adb', 'shell', 'getprop'],
          shell=False,
          stderr=None,
          timeout=adb.DEFAULT_GETPROP_TIMEOUT_SEC,
      )
      self.assertEqual(mock_sleep.call_count, 2)
      mock_sleep.assert_called_with(1)

  def test_forward(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      adb.AdbProxy().forward(['tcp:12345', 'tcp:98765'])
      mock_exec_cmd.assert_called_with(
          ['adb', 'forward', 'tcp:12345', 'tcp:98765'],
          shell=False,
          timeout=None,
          stderr=None,
      )

  def test_reverse(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      adb.AdbProxy().reverse(['tcp:12345', 'tcp:98765'])
      mock_exec_cmd.assert_called_with(
          ['adb', 'reverse', 'tcp:12345', 'tcp:98765'],
          shell=False,
          timeout=None,
          stderr=None,
      )

  def test_instrument_without_parameters(self):
    """Verifies the AndroidDevice object's instrument command is correct in
    the basic case.
    """
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      output = adb.AdbProxy().instrument(MOCK_INSTRUMENTATION_PACKAGE)
      mock_exec_cmd.assert_called_once_with(
          ['adb', 'shell', MOCK_BASIC_INSTRUMENTATION_COMMAND],
          shell=False,
          timeout=None,
          stderr=None,
      )
      self.assertEqual(output, mock_exec_cmd.return_value)

  def test_instrument_with_runner(self):
    """Verifies the AndroidDevice object's instrument command is correct
    with a runner specified.
    """
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      stdout = adb.AdbProxy().instrument(
          MOCK_INSTRUMENTATION_PACKAGE, runner=MOCK_INSTRUMENTATION_RUNNER
      )
      mock_exec_cmd.assert_called_once_with(
          ['adb', 'shell', MOCK_RUNNER_INSTRUMENTATION_COMMAND],
          shell=False,
          timeout=None,
          stderr=None,
      )
      self.assertEqual(stdout, mock_exec_cmd.return_value)

  def test_instrument_with_options(self):
    """Verifies the AndroidDevice object's instrument command is correct
    with options.
    """
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      stdout = adb.AdbProxy().instrument(
          MOCK_INSTRUMENTATION_PACKAGE, options=MOCK_INSTRUMENTATION_OPTIONS
      )
      mock_exec_cmd.assert_called_once_with(
          ['adb', 'shell', MOCK_OPTIONS_INSTRUMENTATION_COMMAND],
          shell=False,
          timeout=None,
          stderr=None,
      )
      self.assertEqual(stdout, mock_exec_cmd.return_value)

  def test_instrument_with_handler(self):
    """Verifies the AndroidDevice object's instrument command is correct
    with a handler passed in.
    """

    def mock_handler(raw_line):
      pass

    with mock.patch.object(
        adb.AdbProxy, '_execute_and_process_stdout'
    ) as mock_execute_and_process_stdout:
      stderr = adb.AdbProxy().instrument(
          MOCK_INSTRUMENTATION_PACKAGE, handler=mock_handler
      )
      mock_execute_and_process_stdout.assert_called_once_with(
          ['adb', 'shell', MOCK_BASIC_INSTRUMENTATION_COMMAND],
          shell=False,
          handler=mock_handler,
      )
      self.assertEqual(stderr, mock_execute_and_process_stdout.return_value)

  def test_instrument_with_handler_with_runner(self):
    """Verifies the AndroidDevice object's instrument command is correct
    with a handler passed in and a runner specified.
    """

    def mock_handler(raw_line):
      pass

    with mock.patch.object(
        adb.AdbProxy, '_execute_and_process_stdout'
    ) as mock_execute_and_process_stdout:
      stderr = adb.AdbProxy().instrument(
          MOCK_INSTRUMENTATION_PACKAGE,
          runner=MOCK_INSTRUMENTATION_RUNNER,
          handler=mock_handler,
      )
      mock_execute_and_process_stdout.assert_called_once_with(
          ['adb', 'shell', MOCK_RUNNER_INSTRUMENTATION_COMMAND],
          shell=False,
          handler=mock_handler,
      )
      self.assertEqual(stderr, mock_execute_and_process_stdout.return_value)

  def test_instrument_with_handler_with_options(self):
    """Verifies the AndroidDevice object's instrument command is correct
    with a handler passed in and options.
    """

    def mock_handler(raw_line):
      pass

    with mock.patch.object(
        adb.AdbProxy, '_execute_and_process_stdout'
    ) as mock_execute_and_process_stdout:
      stderr = adb.AdbProxy().instrument(
          MOCK_INSTRUMENTATION_PACKAGE,
          options=MOCK_INSTRUMENTATION_OPTIONS,
          handler=mock_handler,
      )
      mock_execute_and_process_stdout.assert_called_once_with(
          ['adb', 'shell', MOCK_OPTIONS_INSTRUMENTATION_COMMAND],
          shell=False,
          handler=mock_handler,
      )
      self.assertEqual(stderr, mock_execute_and_process_stdout.return_value)

  @mock.patch.object(adb.AdbProxy, '_exec_cmd')
  def test_root_success(self, mock_exec_cmd):
    mock_exec_cmd.return_value = MOCK_ROOT_SUCCESS_OUTPUT
    output = adb.AdbProxy().root()
    mock_exec_cmd.assert_called_once_with(
        ['adb', 'root'], shell=False, timeout=None, stderr=None
    )
    self.assertEqual(output, MOCK_ROOT_SUCCESS_OUTPUT)

  @mock.patch('time.sleep', return_value=mock.MagicMock())
  @mock.patch.object(adb.AdbProxy, '_exec_cmd')
  def test_root_success_with_retry(self, mock_exec_cmd, mock_sleep):
    mock_exec_cmd.side_effect = [
        adb.AdbError('adb root', '', MOCK_ROOT_ERROR_OUTPUT, 1),
        MOCK_ROOT_SUCCESS_OUTPUT,
    ]
    output = adb.AdbProxy().root()
    mock_exec_cmd.assert_called_with(
        ['adb', 'root'], shell=False, timeout=None, stderr=None
    )
    self.assertEqual(output, MOCK_ROOT_SUCCESS_OUTPUT)
    self.assertEqual(mock_sleep.call_count, 1)
    mock_sleep.assert_called_with(10)

  @mock.patch('time.sleep', return_value=mock.MagicMock())
  @mock.patch.object(adb.AdbProxy, '_exec_cmd')
  def test_root_raises_adb_error_when_all_retries_failed(
      self, mock_exec_cmd, mock_sleep
  ):
    mock_exec_cmd.side_effect = adb.AdbError(
        'adb root', '', MOCK_ROOT_ERROR_OUTPUT, 1
    )
    expected_msg = (
        'Error executing adb cmd "adb root". ret: 1, stdout: , stderr: %s'
        % MOCK_ROOT_ERROR_OUTPUT
    )
    with self.assertRaisesRegex(adb.AdbError, expected_msg):
      adb.AdbProxy().root()
      mock_exec_cmd.assert_called_with(
          ['adb', 'root'], shell=False, timeout=None, stderr=None
      )
    self.assertEqual(mock_sleep.call_count, adb.ADB_ROOT_RETRY_ATTEMPTS - 1)
    mock_sleep.assert_has_calls([mock.call(10), mock.call(20)])

  def test_has_shell_command_called_correctly(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT
      adb.AdbProxy().has_shell_command(MOCK_SHELL_COMMAND)
      mock_exec_cmd.assert_called_once_with(
          ['adb', 'shell', 'command', '-v', MOCK_SHELL_COMMAND],
          shell=False,
          timeout=None,
          stderr=None,
      )

  def test_has_shell_command_with_existing_command(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = MOCK_COMMAND_OUTPUT
      self.assertTrue(adb.AdbProxy().has_shell_command(MOCK_SHELL_COMMAND))

  def test_has_shell_command_with_missing_command_on_older_devices(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT
      mock_exec_cmd.side_effect = adb.AdbError(
          MOCK_ADB_SHELL_COMMAND_CHECK, '', '', 0
      )
      self.assertFalse(adb.AdbProxy().has_shell_command(MOCK_SHELL_COMMAND))

  def test_has_shell_command_with_missing_command_on_newer_devices(self):
    with mock.patch.object(adb.AdbProxy, '_exec_cmd') as mock_exec_cmd:
      mock_exec_cmd.return_value = MOCK_DEFAULT_COMMAND_OUTPUT
      mock_exec_cmd.side_effect = adb.AdbError(
          MOCK_ADB_SHELL_COMMAND_CHECK, '', '', 1
      )
      self.assertFalse(adb.AdbProxy().has_shell_command(MOCK_SHELL_COMMAND))

  @mock.patch.object(adb.AdbProxy, 'getprop')
  @mock.patch.object(adb.AdbProxy, '_exec_cmd')
  def test_current_user_id_25_and_above(self, mock_exec_cmd, mock_getprop):
    mock_getprop.return_value = b'25'
    mock_exec_cmd.return_value = b'123'
    user_id = adb.AdbProxy().current_user_id
    mock_exec_cmd.assert_called_once_with(
        ['adb', 'shell', 'am', 'get-current-user'],
        shell=False,
        stderr=None,
        timeout=None,
    )
    self.assertEqual(user_id, 123)

  @mock.patch.object(adb.AdbProxy, 'getprop')
  @mock.patch.object(adb.AdbProxy, '_exec_cmd')
  def test_current_user_id_between_21_and_24(self, mock_exec_cmd, mock_getprop):
    mock_getprop.return_value = b'23'
    mock_exec_cmd.return_value = (
        b'Users:\n'
        b'UserInfo{123:Owner:13} serialNo=0\n'
        b'Created: <unknown>\n'
        b'Last logged in: +1h22m12s497ms ago\n'
        b'UserInfo{456:Owner:14} serialNo=0\n'
        b'Created: <unknown>\n'
        b'Last logged in: +1h01m12s497ms ago\n'
    )
    user_id = adb.AdbProxy().current_user_id
    mock_exec_cmd.assert_called_once_with(
        ['adb', 'shell', 'dumpsys', 'user'],
        shell=False,
        stderr=None,
        timeout=None,
    )
    self.assertEqual(user_id, 123)


if __name__ == '__main__':
  unittest.main()
